package com.flink.examples.file;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.util.Collector;
import scala.Tuple7;

import java.util.Date;

/**
 * @Description 从txt文件中读取内容输出到DataSet中
 * @Author JL
 * @Date 2020/09/18
 * @Version V1.0
 */
public class TextSource {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String filePath = "D:\\Workspaces\\idea_2\\flink-examples\\connectors\\src\\main\\resources\\user.txt";
        DataSet<Tuple7<Integer, String, Integer, Integer, String, Date, Long>> dataSet = env.readTextFile(filePath)
                .flatMap(new FlatMapFunction<String, Tuple7<Integer, String, Integer, Integer, String, Date, Long>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple7<Integer, String, Integer, Integer, String, Date, Long>> out) throws Exception {
                        if (StringUtils.isNotBlank(value)) {
                            String[] values = value.split(",");
                            out.collect(Tuple7.apply(
                                    Integer.parseInt(values[0]),
                                    values[1],
                                    Integer.parseInt(values[2]),
                                    Integer.parseInt(values[3]),
                                    values[4],
                                    DateUtils.parseDate(values[5], "yyyy-MM-dd HH:mm:ss"),
                                    Long.parseLong(values[6])));
                        }
                    }
                });
        dataSet.print();
    }

}
/*
(13,liu1,21,1,USA,Sun Sep 06 00:00:00 CST 2020,1599321600000)
(8,liu3,21,1,CN,Tue Sep 01 00:00:00 CST 2020,1598889600000)
(9,wang1,23,1,CN,Wed Sep 02 00:00:00 CST 2020,1598976000000)
 */